package com.cssl;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

/**
 * @Author: chen
 * @Date: 2021/11/10 23:02
 * @Desc: flink-cdc table api
 */
public class FlinkCDCWithSql {
    public static void main(String[] args) throws Exception {
        System.setProperty("HADOOP_USER_NAME", "atguigu");
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        tableEnv.executeSql("CREATE TABLE base_trademark (" +
                " id STRING NOT NULL," +
                " tm_name STRING," +
                " logo_url STRING" +
                ") WITH (" +
                " 'connector' = 'mysql-cdc'," +
                " 'hostname' = 'hadoop102'," +
                " 'port' = '3306'," +
                " 'username' = 'root'," +
                " 'password' = 'root'," +
                " 'database-name' = 'gmall-flink'," +
                " 'table-name' = 'base_trademark'" +
                ")");

        Table table = tableEnv.sqlQuery("select * from base_trademark");

        tableEnv.toRetractStream(table, Row.class).print();

        env.execute("flink-cdc-sql");
    }
}
